package PrjWsMQ

import (
	"bytes"
	"dgo/goutils/utils"
	"fmt"
	"strings"
)

type WsMQClt struct {
	ID                 string
	scribleChannelsMap utils.GSyncMapGroup
	ws                 *WsClient
	onRecvTopic        func(sender *WsMQClt, topic string, msg []byte)
}

func NewWsMQClt(ws_req_url, origin_url string) *WsMQClt {
	if len(origin_url) == 0 {
		protocol := "http://"
		tmpUrl := ws_req_url
		if strings.HasPrefix(tmpUrl, "ws://") {
			tmpUrl = strings.TrimPrefix(tmpUrl, "ws://")
		} else if strings.HasPrefix(tmpUrl, "wss://") {
			tmpUrl = strings.TrimPrefix(tmpUrl, "wss://")
			protocol = "https://"
		}
		strs := strings.SplitN(tmpUrl, "/", 2)
		origin_url = protocol + strs[0]
	}
	rval := &WsMQClt{ws: NewWsClient(ws_req_url, origin_url)}
	rval.ws.OnConnect = rval.onWsConnected
	rval.ws.OnRecv = rval.onWsRecv
	return rval
}

func (this *WsMQClt) ConfigOnRecvTopic(onRecvTopic func(sender *WsMQClt, topic string, msg []byte)) {
	this.onRecvTopic = onRecvTopic
}

func (this *WsMQClt) Start() {
	this.ws.Start()

}

func (this *WsMQClt) Stop() {
	this.ws.Stop()
}

func (this *WsMQClt) PublishMessage(topic string, msg string) error {
	//var bb utils.BytesBuilder
	//bb.AppendStr(topic).AppendStr(",")
	//bb.Write(msg)
	//return this.ws.SendMsg(WS_MESSAGE_TEXT, bb.Bytes())

	return this.ws.SendMsg(WS_MESSAGE_TEXT, []byte(fmt.Sprintf("%s,%s", topic, msg)))
}

func (this *WsMQClt) PublishBuf(topic string, msg []byte) error {
	var bb utils.BytesBuilder
	bb.AppendStr(topic).Write(buf_spliter)
	bb.Write(msg)
	return this.ws.SendMsg(WS_MESSAGE_BINARY, bb.Bytes())
}

func (this *WsMQClt) PublishStrBuf(topic string, msg []byte) error {
	var bb utils.BytesBuilder
	bb.AppendStr(topic).AppendStr(",")
	bb.Write(msg)
	return this.ws.SendMsg(WS_MESSAGE_TEXT, bb.Bytes())
}

func (this *WsMQClt) onWsRecv(sender *WsClient, msgtype byte, msg []byte) {
	if this.onRecvTopic == nil {
		return
	}
	if msgtype == WS_MESSAGE_TEXT {
		str := string(msg)
		strs := strings.SplitN(str, ",", 2)
		if len(strs) > 1 {
			this.onRecvTopic(this, strs[0], []byte(strs[1]))
		}
	} else if msgtype == WS_MESSAGE_BINARY { // BinaryMessage
		nbuf := bytes.SplitN(msg, buf_spliter, 2)
		id := string(nbuf[0])
		if len(nbuf) > 1 {
			// 发送源数据, 带有ChannelID
			this.onRecvTopic(this, id, nbuf[1])
		}
	}

}

/*
*
用逗号分隔
*/
func (this *WsMQClt) SubScribleChannel(channels string) {
	strs := strings.Split(channels, ",")
	for _, v := range strs {
		channelid := utils.Trim(v)
		if len(channelid) == 0 {
			continue
		}
		_, loaded := this.scribleChannelsMap.LoadOrStoreFunc2(channelid, func() interface{} {
			return channelid
		}, nil)
		if !loaded {
			if this.ws.IsConnected() {
				this.ws.SendMsg(WS_MESSAGE_TEXT, []byte(fmt.Sprintf("/cmd,subscrible,%s", channelid)))
			}
		}
	}
}

// 服务端尚未实现
func (this *WsMQClt) UnsubScribleChannel(channels string) {
	strs := strings.Split(channels, ",")
	for _, v := range strs {
		channelid := utils.Trim(v)
		if len(channelid) == 0 {
			continue
		}
		if this.ws.IsConnected() {

			this.ws.SendMsg(WS_MESSAGE_TEXT, []byte(fmt.Sprintf("/cmd,unsubscrible,%s", channelid)))
		}
		this.scribleChannelsMap.Remove(channelid)
	}
}

func (this *WsMQClt) onWsConnected(sender *WsClient) {
	channels := this.scribleChannelsMap.ValAsString2(",", nil, 0)
	err := this.ws.SendMsg(WS_MESSAGE_TEXT, []byte(fmt.Sprintf("/cmd,subscrible,%s", channels)))
	if err != nil {
		utils.Log.WarnTypef("STATE", "[%s]请求订阅主题(%s)失败:%s", this.ID, channels, err.Error())
	} else {
		utils.Log.InfoTypef("STATE", "[%s]请求订阅主题(%s)成功", this.ID, channels)
	}

}
